package com.test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.ConnectionFactory;

import reactor.core.Disposable;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Receiver;
import reactor.rabbitmq.ReceiverOptions;

public class Consumer {

	private Logger LOGGER = LoggerFactory.getLogger(Consumer.class);

	@Test
    public void testConsumer() throws InterruptedException {
        int count = 2;
        CountDownLatch latch = new CountDownLatch(count);

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.useNio();
        connectionFactory.setVirtualHost("/sms");
        connectionFactory.setUsername("sms_gaea-servicer-provider");
        connectionFactory.setPassword("dGGXPlY6YO");

        ReceiverOptions receiverOptions = new ReceiverOptions()
                .connectionFactory(connectionFactory)
                .connectionSupplier(cf -> cf.newConnection(
                		new Address[] {new Address("10.216.4.233",5672), new Address("10.216.4.234",5672), new Address("10.216.4.236",5672)},
                        "reactive-receiver"))
                .connectionSubscriptionScheduler(Schedulers.elastic());

        Receiver receiver = RabbitFlux.createReceiver(receiverOptions);
        Disposable disposable = receiver.consumeAutoAck("sms.q.alevel")
        .subscribe(m -> {
            LOGGER.info("Received message {}", new String(m.getBody()));
            latch.countDown();
        });

        latch.await(10, TimeUnit.SECONDS);

        disposable.dispose();
        receiver.close();
    }

}
